package com.lg.flink.db;

import com.lg.flink.bean.Order;
import com.lg.flink.bean.WC;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static com.lg.flink.db.DbConfig.*;

public class DbCountTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Row> dataStreamSource = dataSource(env);
        dataStreamSource.
                flatMap(new FlatMapFunction<Row, WC>() {
                    @Override
                    public void flatMap(Row value, Collector<WC> out) throws Exception {
                        Tuple2<String, Integer> tuple2 = new Tuple2<String, Integer>();
                        tuple2.setField(value.getField(1), 0);
                        tuple2.setField(1, 1);
                        WC wc = new WC();
                        wc.setWord((String) value.getField(1));
                        wc.setFrequency(1);
                        out.collect(wc);
                    }
                })
                .keyBy("word")
                .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum("frequency")
                .print();
        env.execute("Window WordCount");
    }

    private static DataStreamSource<Row> dataSource(StreamExecutionEnvironment env) {
        JDBCInputFormat inputBuilder = JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(driverClass)
                .setDBUrl(dbUrl)
                .setUsername(userNmae)
                .setPassword(passWord)
                .setQuery("select * from bi_behivor")
                // 这里第一个字段类型写int会报类型转换异常。
                .setRowTypeInfo(new RowTypeInfo(
                        BasicTypeInfo.LONG_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.DATE_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO)).finish();

        DataStreamSource<Row> source = env.createInput(inputBuilder);
        return source;
    }
}
